#pragma once

#include <list>
#include <queue>
#include <mutex>
#include <shared_mutex>

#include <bvar/bvar.h>

#include "hawking/indexlib/schema.pb.h"
#include "hawking/indexlib/indexlib_data.pb.h"

#include "doc/document_creator.h"
#include "segment/disk_segment.h"
#include "segment/mem_segment.h"
#include "util/file_wrapper/file_wrapper.h"

namespace bvar {
extern PassiveStatus<long> g_resident;
};

inline long GetMetricOfProcessMemoryResident() {
    return bvar::g_resident.get_value() / (1024.0 * 1024.0);
}

namespace hawking {
namespace indexlib {

class DiskSegmentMergeStrategy {
public:
    DiskSegmentMergeStrategy(
        std::list<std::shared_ptr<DiskSegment>> disk_segments) :
        disk_segments_(disk_segments) {}
    virtual ~DiskSegmentMergeStrategy() = default;

    virtual std::list<std::shared_ptr<DiskSegment>> FindMergeWorkItems() = 0;

protected:
    std::list<std::shared_ptr<DiskSegment>> disk_segments_;
};

// 1. prefer merge smaller segments to one new segment as could(memory limit)
// 2. urgent segments(ShouldDump return true) is prefered
class SimpleDiskSegmentMergeStrategy : public DiskSegmentMergeStrategy {
public:
    SimpleDiskSegmentMergeStrategy(
        std::list<std::shared_ptr<DiskSegment>> disk_segments) :
            DiskSegmentMergeStrategy(disk_segments) {
        for (const auto disk_segment : disk_segments_) {
            if (disk_segment->ShouldDump()) {
                with_urgent_ = true;
                break;
            }
        }
    }
    virtual ~SimpleDiskSegmentMergeStrategy() = default;

    virtual std::list<std::shared_ptr<DiskSegment>> FindMergeWorkItems() override;

private:
    bool with_urgent_ = false;
};

class Table {
public:
    struct AllSegment {
        std::shared_ptr<IndexlibIndexConfig> schema;
        std::shared_ptr<MemSegment> building_segment;
        std::list<std::shared_ptr<MemSegment>> dumping_segment;
        std::list<std::shared_ptr<DiskSegment>> disk_segment;
        AllSegment(
            std::shared_ptr<IndexlibIndexConfig> config,
            std::shared_ptr<MemSegment> build,
            std::list<std::shared_ptr<MemSegment>> dump,
            std::list<std::shared_ptr<DiskSegment>> disk) :
            schema(config),
            building_segment(build), dumping_segment(dump), disk_segment(disk) {}
    };

    Table(const std::string& table_name, const std::string& root_dir)
        : table_name_(table_name), root_dir_(root_dir), running_(0) {
        segment_idx_.store(0, std::memory_order_relaxed);
    }
    ~Table();

    // use in format
    bool Init(std::shared_ptr<IndexlibIndexConfig> schema);
    // use in both server and format
    bool SnapShot(const std::string& root_dir, bool is_format = true);

    // use in server
    bool Load(
        std::shared_ptr<IndexlibIndexConfig> schema,
        bool need_increase,
        bool use_disk);

    void Query(const IndexlibRetrieveRequest& req, IndexlibRetrieveResponse* resp);
    void QueryByPk(const IndexlibRetrieveByPkRequest& req, IndexlibRetrieveResponse* resp);
    bool FilterCheck(
        int64_t pk,
        const google::protobuf::RepeatedPtrField<util::FilterColumn>& filter_fields,
        const google::protobuf::RepeatedPtrField<util::FilterColumn>& where_fields);
    bool FilterCheck(
        int64_t pk,
        const util::FilterColumn& field,
        bool is_where);
    
    void FillLocaldata(
        int64_t pk,
        google::protobuf::RepeatedPtrField<util::SelectColumn>* select_fields);
    void FillLocaldata(
        int64_t pk,
        util::SelectColumn* select_field);
    void FillLocaldata(
        int64_t pk,
        int32_t field_id,
        util::FieldValue* field_value);

    util::HawkingRecallRetCode AddDocument(const IndexlibMessage& message);

    bool UpdateSchema(std::shared_ptr<IndexlibIndexConfig> schema);

    const std::string& TableName() const {
        return table_name_;
    }

    const std::string& TableDir() const {
        return root_dir_;
    }

    std::shared_ptr<IndexlibIndexConfig> IndexConfig() {
        std::shared_lock<std::shared_mutex> r_lock(mgr_lock_);
        return schema_;
    }

private:
    void Init_();
    util::HawkingRecallRetCode RemoveDocByPk_(uint64_t pk);
    util::HawkingRecallRetCode Table::RemoveDocByPk_(
        std::shared_ptr<Segment> segment, uint64_t pk);
    util::HawkingRecallRetCode AddDocument_(const IndexlibMessage& message, uint64_t pk);
    bool FetchPk_(const IndexlibMessage& message, uint64_t* pk) const;
    void InitPkFieldName_() {
        pk_field_name_ = schema_->pk_field_name();
    }
    bool FetchTerm_(
        std::shared_ptr<IndexlibIndexConfig> schema,
        const IndexlibRetrieveRequest& req,
        uint64_t* term);

    ////////////////////////////////////////////////////// index config and segment relate ////////
    AllSegment GetAllSegment_() {
        std::shared_lock<std::shared_mutex> r_lock(mgr_lock_);
        return {schema_, building_segment_, dumping_segments_, disk_segments_};
    }

    void UpdateBuildingSegment_(
        std::shared_ptr<MemSegment> segment_will_dump,
        std::shared_ptr<MemSegment> new_building_segment) {
        segment_will_dump->ToDumpingQueue();
        dumping_queue_.push(segment_will_dump);
        dumping_segments_.push_back(segment_will_dump);
        building_segment_ = new_building_segment;
    }

    void UpdateBuildingSegment_(
        std::shared_ptr<MemSegment> segment_will_dump,
        std::shared_ptr<MemSegment> new_building_segment,
        bool with_lock) {
        if (with_lock) {
            std::unique_lock<std::shared_mutex> w_lock(mgr_lock_);
            UpdateBuildingSegment_(segment_will_dump, new_building_segment);
        } else {
            UpdateBuildingSegment_(segment_will_dump, new_building_segment);
        }
    }

    bool LoadDiskSegment_(
        const std::string disk_segment_dir,
        std::shared_ptr<MemSegment> remove_mem_segment = nullptr);

    bool PkExists_(
        uint64_t pk, std::shared_ptr<Segment>* exist_segment, DocId* exist_doc_id) const;
    
    std::string CreateSegmentDir_(const std::string& root_dir) const {
        int32_t unique_idx = segment_idx_.load(std::memory_order_acquire);
        char time_buf[32] = {'\0'};
        ::time_t t = time(nullptr);
        ::strftime(time_buf, sizeof(time_buf), "%Y%m%d%H%M%S", ::localtime(&t));
        std::string segment_dir = butil::string_printf(
            "%s/%s-%s.%d", root_dir.data(), table_name_.data(), time_buf, unique_idx);
        FslibWrapper::RmDir(segment_dir);
        FslibWrapper::Mkdir(segment_dir);
    }

    bool FieldCompletion_(
        IndexlibMessage* mut_message,
        uint64_t pk,
        std::shared_ptr<Segment> exist_segment);

    bool CheckSchema_(std::shared_ptr<IndexlibIndexConfig> schema) const;
    bool DiffExistSchema_(std::shared_ptr<IndexlibIndexConfig> schema) const;
    bool SchemaBasicCheck_(std::shared_ptr<IndexlibIndexConfig> schema) const;

    bool CouldMerge_(
        const std::vector<std::shared_ptr<IndexlibIndexConfig>>& schemas) const;

    FSFileType DecideFsFileTypeAfterMerge_(
        const std::list<std::shared_ptr<DiskSegment>>& merged_disk_segment_list) const;
    ///////////////////////////////////////////////////////////////////////////////////
    std::shared_mutex snapshot_lock_;

    bthread::Mutex data_lock_;
    std::shared_mutex mgr_lock_;

    bool use_disk_ = false;
    std::string pk_field_name_ = "";
    std::string table_name_ = "";
    std::string root_dir_ = "";
    std::atomic<int32_t> running_;

    std::mutex dumping_mutex_;
    std::queue<std::shared_ptr<MemSegment>> dumping_queue_;

    std::mutex garbage_mutex_;
    std::queue<std::shared_ptr<Segment>> garbage_queue_;

    std::mutex merging_mutex_;
    std::queue<std::vector<std::shared_ptr<DiskSegment>>> merging_queue_;

    std::list<std::shared_ptr<DiskSegment>> disk_segments_;
    std::list<std::shared_ptr<MemSegment>> dumping_segments_;
    std::shared_ptr<MemSegment> building_segment_;
    std::thread dumping_thread_;
    std::thread garbage_thread_;
    std::thread merging_thread_;

    // support config update
    std::shared_ptr<IndexlibIndexConfig> schema_;

    // use for format
    bool building_in_one_segment_ = false;

    // segment_idx_ has no special meaning. only a auto-incr,
    // only for absolutely distinguish different segment with timestmap.
    std::atomic<int32_t> segment_idx_;
};

}
}